/*
 *    Copyright 2022 The DSMS Authors.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

package com.dsms.modules.task.service.impl;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.dsms.common.constant.TaskStatusEnum;
import com.dsms.common.taskmanager.model.Task;
import com.dsms.common.taskmanager.service.ITaskService;
import com.dsms.dfsbroker.cluster.model.Cluster;
import com.dsms.dfsbroker.cluster.service.IClusterService;
import com.dsms.modules.server.CommonUpdateServer;
import com.dsms.modules.task.mapper.TaskMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

/**
 * Implementation of task operation service
 */
@Service
@Slf4j
public class TaskServiceImpl extends ServiceImpl<TaskMapper, Task> implements ITaskService {

    @Autowired
    private IClusterService clusterService;

    @Override
    public void updateFrontTaskInfo(Task task) {
        if (ObjectUtils.isEmpty(task)) {
            return;
        }
        //when task's status become "FINISH" or "FAIL" means it has been done,and we should notify the front
        //specially,when task's status become "ROLLBACK" means it has been failed
        if (!(task.getTaskStatus().equals(TaskStatusEnum.FINISH.getStatus())
                || task.getTaskStatus().equals(TaskStatusEnum.FAIL.getStatus())
                || task.getTaskStatus().equals(TaskStatusEnum.ROLLBACK.getStatus())
                || task.getTaskStatus().equals(TaskStatusEnum.ROLLBACK_SUCCESS.getStatus()))) {
            return;
        }

        CommonUpdateServer.sendMessageToAll(JSON.toJSONString(task));
    }

    @Override
    public boolean validateBindCluster() {
        boolean isBindCluster = false;
        try {
            Cluster currentBindCluster = clusterService.getCurrentBindCluster();
            if (!ObjectUtils.isEmpty(currentBindCluster)) {
                isBindCluster = true;
            }
        } catch (Exception e) {
            log.warn("validate Bind Cluster error.error message:{}", e.getMessage());
            return false;
        }
        return isBindCluster;

    }

    @Override
    public boolean validateTaskMessageAndTaskType(String taskMessage) {
        LambdaQueryWrapper<Task> taskQuery = new LambdaQueryWrapper<>();
        taskQuery.in(Task::getTaskStatus, TaskStatusEnum.QUEUE.getStatus(), TaskStatusEnum.EXECUTING.getStatus())
                .eq(Task::getTaskMessage, taskMessage);
        long count = this.count(taskQuery);

        return count <= 0;
    }

}
